package distributeCache;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * @author legolas
 * @date 2020/3/19 下午1:58
 * 将权重指标文件，当作分布式缓存文件导入缓存，统计文件与权重指标相乘，得到最终统计结果
 * 缓存文件：weight.json：{"WA":0.2,"WB":0.3,"WC":0.2,"WD":0.3}
 * 要素文件：forecast.json：{"A":12,"B":23,"C":54,"D":32}
 */
public class StatisticalTestJob {
    public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

        private Map<String, String> cacheMap = new HashMap<>();

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            super.setup(context);

            //FileReader reader = new FileReader("mycache");
            //BufferedReader br = new BufferedReader(reader);
            JSONObject jb = JSON.parseObject("{\"A\":12,\"B\":23,\"C\":54,\"D\":32}");

            cacheMap.put("ID", jb.getString("ID"));
            cacheMap.put("WA", jb.getString("WA"));
            cacheMap.put("WB", jb.getString("WB"));
            cacheMap.put("WC", jb.getString("WC"));
            cacheMap.put("WD", jb.getString("WD"));

        }

        /**
         * 需要实现的map函数
         * 这个map函数就是可以接收k1，v1   产生k2，v2
         *
         * @param k1
         * @param v1
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */


        @Override
        protected void map(LongWritable k1, Text v1, Context context) throws IOException, InterruptedException {
            //k1代表的是每一行的行首偏移量，v1代表的时每一行的内容
            //对获取到的每一行数据进行切割，把单词切割出来
            String line = v1.toString();
            JSONObject jsonObj = JSON.parseObject(line);

            String ID = jsonObj.getString("ID");
            long A = jsonObj.getLongValue("A");
            long B = jsonObj.getLongValue("B");
            long C = jsonObj.getLongValue("C");
            long D = jsonObj.getLongValue("D");
            //迭代切割出来的单词数据
            Long sum = 0L;

            sum += Long.parseLong(cacheMap.get("WA")) * A;
            sum += Long.parseLong(cacheMap.get("WB")) * B;
            sum += Long.parseLong(cacheMap.get("WC")) * C;
            sum += Long.parseLong(cacheMap.get("WD")) * D;
            Text k2 = new Text();
            k2.set(ID);
            LongWritable v2 = new LongWritable();
            v2.set(sum);
            context.write(k2, v2);
        }
    }


    /**
     * 组装job： map+reduce
     *
     * @param args
     */
    public static void main(String[] args) {
        try {
            if (args.length != 2) {
                //如果参数不够，程序直接退出
                System.out.print("请指定输入路径和输出路径！");
                System.exit(100);
            }
            //创建Job需要的配置参数
            Configuration conf = new Configuration();

            //创建一个job
            Job job = Job.getInstance(conf);

            //String cache = "hdfs://hadoopmaster:9000/statistic/cache/weight.json";//目录或文件
            //cache = cache + "#mycache"; //mycache为缓存别名
            // job.addCacheArchive(new URI(cache));//添加到job设置

            job.setJarByClass(distributeCache.StatisticalTestJob.class);

            //指定输入路径
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            //指定map相关代码
            job.setMapperClass(MyMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);

            //指定reduce相关代码
            job.setNumReduceTasks(0);

            //提交job
            job.waitForCompletion(true);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
